package com.thrift.server.rpc.server;

import com.thrift.server.rpc.impl.ServerServiceImpl;
import com.thrift.server.rpc.thriftcode.StudentService;
import org.apache.commons.lang.StringUtils;
import org.apache.thrift.TMultiplexedProcessor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.server.THsHaServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TTransportException;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;

import javax.annotation.Resource;
import java.net.InetAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author 何锦涛--CodeSleep
 * @since 2022/9/9 22:38
 */
public class ThriftServer {

    private static final Logger logger = LoggerFactory.getLogger(TServerZk.class);
    // RPC 所有相关节点公用的顶级父节点
    private static final String THRIFT_SERVER_PREFIX = "/thrift/";
    // 固定的一个单线程池
    private final ExecutorService thread = Executors.newSingleThreadExecutor();
    // RPC 服务名
    private final String name;
    // RPC 服务端口号
    private final int port;
    // zk 服务实例
    private final TServerZk serverZk;
    // 线程池最小线程数
    @Value("${server.thrift.min-thread-pool}")
    private final Integer minThreadPool = 100;
    // 线程池最大线程数
    @Value("${server.thrift.max-thread-pool}")
    private final Integer maxThreadPool = 1000;
    // 业务服务对象
    @Resource
    ServerServiceImpl myServerService;

    /**
     * 构造函数注册到 zk 中并启动 RPC 节点
     *
     * @param port     RPC 端口号
     * @param name     RPC 服务名
     * @param serverZk zk 服务实例
     */
    public ThriftServer(int port, String name, TServerZk serverZk) {
        this.serverZk = serverZk;
        this.port = port;
        this.name = name;
        this.registerService();
    }

    /**
     * 构造函数注册到 zk 中并启动 RPC 节点
     */
    private void startAndRegisterService() {
        if (!start()) {
            logger.error("启动 RPC server 节点错误");
        }
        registerService();
    }

    /**
     * 启动 RPC 服务
     * @return 是否启动成功
     */
    private boolean start() {
        try {
            TMultiplexedProcessor processor = new TMultiplexedProcessor();
            TNonblockingServerSocket socket = new TNonblockingServerSocket(port);
            THsHaServer.Args arg = new THsHaServer.Args(socket).minWorkerThreads(minThreadPool).maxWorkerThreads(maxThreadPool);
            arg.protocolFactory(new TBinaryProtocol.Factory());
            // serviceName 使用 this.name
            processor.registerProcessor(this.name, new StudentService.Processor<StudentService.Iface>(myServerService));
            arg.processor(processor);
            TServer server = new THsHaServer(arg);
            logger.info("启动 thrift server; port:" + port);
            // 异步提交线程，防止主线程阻塞
            thread.submit(server::serve);
            int count = 0;
            while (!server.isServing()) {
                logger.error("等待启动 thrift server，轮训次数：{}", ++count);
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    logger.error("等待启动 thrift server 轮训发生线程异常");
                }
            }
        } catch (TTransportException e) {
            logger.error("启动 RPC 服务错误：{}", e.toString());
            return false;
        }
        return true;
    }

    /**
     * 注册此 RPC 服务节点；注册完之后就是 /thrift/xxx/10.1.38.226:7778 临时节点
     */
    private void registerService() {
        // 判断是否存在父节点
        if (!serverZk.exist(THRIFT_SERVER_PREFIX.concat(name))) {
            register();
        }
        String serviceNodePath = getServiceNodePath();
        if (StringUtils.isBlank(serviceNodePath)) {
            return;
        }
        // 如果已经存在该 RPC 服务节点，就不用再创建了
//        if (!serverZk.exist(THRIFT_SERVER_PREFIX.concat(name).concat(serviceNodePath))) {
//            return;
//        }
        serverZk.add(THRIFT_SERVER_PREFIX.concat(name).concat(serviceNodePath), String.valueOf(port).getBytes(), CreateMode.EPHEMERAL);
    }

    /**
     * 注册 RPC 服务父节点，比如我们的 RPC 服务为 hello-server，注册完之后就是 /thrift/hello-server 永久节点
     */
    private void register() {
        if (!serverZk.exist(THRIFT_SERVER_PREFIX.concat(name))) {
            serverZk.add(THRIFT_SERVER_PREFIX.concat(name), name.getBytes(), CreateMode.PERSISTENT);
        }
    }

    /**
     * 获取本服务即将注册到 ZK 上的节点路径，根据本机器 IP 和定义的端口号生成唯一路径
     */
    private String getServiceNodePath() {
        try {
            // 获取本机 IP
            InetAddress inetAddress = InetAddress.getLocalHost();
            String servicePath = inetAddress.getHostAddress();
            return "/".concat(servicePath).concat(":").concat(String.valueOf(port));
        } catch (Exception e) {
            return null;
        }
    }
}
